package net.lzzy.gsonandjson;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;

public class CarMaxTest {
    public static void main(String[] args) throws Exception {
        SparkConf sparkConf = new SparkConf().setAppName("CarMaxTest").setMaster("local");
        JavaSparkContext javaSparkContext = new JavaSparkContext(sparkConf);
        JavaRDD<String> stringJavaRDD = javaSparkContext.textFile("src/can_data.json");
        JavaRDD<Car> JavaMap = stringJavaRDD.map(new Function<String, Car>() {
            @Override
            public Car call(String s) throws Exception {
                Gson gson = new GsonBuilder().create();
                return gson.fromJson(s, Car.class);
            }
        });
        JavaPairRDD<String, Car> carKey = JavaMap.mapToPair(new PairFunction<Car, String, Car>() {
            @Override
            public Tuple2<String, Car> call(Car car) throws Exception {
                return new Tuple2<>(car.getCollectTime().split(" ")[0], car);
            }
        });
        JavaPairRDD<String, Car> carReduce = carKey.reduceByKey(new Function2<Car, Car, Car>() {
            @Override
            public Car call(Car car, Car car2) throws Exception {
                return car.getSpeed() >= car2.getSpeed() ? car : car2;
            }
        });
        carReduce.collect().forEach(item->{
            System.out.println(item._2);
        });

    }
}
